package voldemort.store.cachestore.impl;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import voldemort.store.cachestore.BlockOps;
import voldemort.store.cachestore.CacheBlock;
import voldemort.store.cachestore.Key;
import voldemort.store.cachestore.StoreException;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;

import static voldemort.store.cachestore.BlockUtil.*;

/**
 * Created by IntelliJ IDEA.
 * User: derek
 * Date: May 15, 2010
 * Time: 1:03:42 PM
 * To change this template use File | Settings | File Templates.
 */
public class ChannelStore implements BlockOps<byte[]> {

    private static Log logger = LogFactory.getLog(ChannelStore.class);
    // when data reside
    private FileChannel dataChannel;
    // key for map reside, it is write once
    private FileChannel keyChannel;
    // fixed len index block for dataChannel and keyChannel
    private FileChannel indexChannel;
    // delete log file
    private FileChannel logChannel;
    // signature of data and index file
    private final int MAGIC = 0xBABECAFE;
    // beginning of index channel
    private final static int OFFSET = 4;
    // total number of record
    private volatile int totalRecord = 0 ;
    // current offset of dataf
    private volatile long dataOffset = 0;
    // key channel offset
    private volatile long keyOffset = 0;
    // logOffset
    private volatile long logOffset = 0;
    // map hold all key in memory and cacheBlock. Data may be, data is cacheable, but key is not
    private final ReentrantLock lock = new ReentrantLock();
    // number of deleted record
    private volatile int deleted;
    // plug in interface to determine blockSize
    //private BlockSize blockSize;
    // map hold all key in memory and cacheBlock. Data may be, data is cacheable, but key is not
    private Map<Key, CacheBlock> map;
    // filename
    private String filename;

    private ChannelStore(String filename, Map map) {
        this(filename, map, false);
    }
    private ChannelStore(String filename, Map map, boolean reset) {
        try {
            this.map = map;
            this.filename = filename ;
            indexChannel = new RandomAccessFile(filename+".ndx", "rw").getChannel();
            checkSignature( indexChannel);
            keyChannel   = new RandomAccessFile(filename+".key", "rw").getChannel();
            checkSignature( keyChannel);
            dataChannel  = new RandomAccessFile(filename+".data", "rw").getChannel();
            checkSignature( dataChannel);
            logChannel  = new RandomAccessFile(filename+".log", "rw").getChannel();
            checkSignature( logChannel);
            init( reset);
        } catch (IOException ex) {
            throw new StoreException(ex.getMessage(), ex);
        }
    }


    public static ChannelStore open(String filename, Map map) {
        return new ChannelStore(filename, map);
    }
    
    public static ChannelStore open(String filename, Map map, boolean reset) {
        return new ChannelStore(filename, map, reset);
    }

    public String getFilename() {
        return filename;
    }

    public static boolean isChannelExist(String filename) {
        File file = new File( filename);
        if ( ! file.isFile() ) return false;
        if ( file.length() >  OFFSET + RECORD_SIZE )
            return true;
        else
            return false;
    }

    private boolean checkSignature(FileChannel channel) throws IOException {
        ByteBuffer intBytes = ByteBuffer.allocate(OFFSET);
        if ( channel.size() == 0) {
           intBytes.putInt(MAGIC);
           intBytes.flip();
           channel.write(intBytes);
        }
        else {
           channel.read(intBytes);
           intBytes.rewind();
           if ( intBytes.getInt() != MAGIC )
               throw new StoreException("Header mismatch expect "+MAGIC+" read "+ intBytes.getInt() );
        }
        return true;
    }

    private void init( boolean reset) throws IOException {
        if ( reset ) {
            indexChannel.truncate( OFFSET);
            dataChannel.truncate( OFFSET);
            keyChannel.truncate( OFFSET);
            logChannel.truncate( OFFSET);
            totalRecord = 0 ;
        }
        else {
            long length = indexChannel.size() - OFFSET ;
            totalRecord = (int) ( length / RECORD_SIZE) ;
            ByteBuffer buf = ByteBuffer.allocate(RECORD_SIZE);
            logger.info("Building key map and read index file for "+filename+" total record " + totalRecord  );
            for ( int i = 0 ; i < totalRecord ; i ++) {
                indexChannel.read(buf);
                assert ( buf.capacity() == RECORD_SIZE);
                buf.rewind();
                byte status = buf.get();
                if ( isDeleted(status) ) this.deleted ++ ;
                else {
                    long key = buf.getLong();
                    byte[] keys = readChannel( key, keyChannel);
                    long data = buf.getLong();
                    long block2version = buf.getLong();
                    CacheBlock block = new CacheBlock(i, data, block2version, status );
                    map.put( toKey( keys) , block);
                }
                buf.clear();
            }
        }
        dataOffset = dataChannel.size();
        keyOffset = keyChannel.size();
        logOffset = logChannel.size();
        logger.info("Total record " + totalRecord + " deleted "+deleted+" active "+ (totalRecord - deleted) );
    }

    private boolean isDeleted(byte b) {
        if ( (b & DELETED) == DELETED ) return true;
        else return false;
    }


    /**
     *
     * @param offset2len
     * @param channel
     * @return
     * @throws IOException
     */
    public byte[] readChannel(long offset2len, FileChannel channel)  throws IOException {
        long offset = getOffset( offset2len);
        int len = getLen( offset2len);
        ByteBuffer data = ByteBuffer.allocate( len);
        channel.read( data, offset);
        return data.array();
    }


    public void writeNewBlock(CacheBlock<byte[]> block, long keyOffset2Len, byte[] key) throws IOException {
        // write in sequence of data, key and index channel, which doen not need call flip()
        ByteBuffer dataBuf = ByteBuffer.wrap( block.getData() );
        dataChannel.write( dataBuf, block.getDataOffset() );
        // position to end of block
        if (block.getBlockSize() > block.getDataLen()  ) {
            ByteBuffer fillBuf = ByteBuffer.wrap( new byte[] {0} );
            dataChannel.write( fillBuf, block.getDataOffset()+ block.getBlockSize()-1 );
        }
        ByteBuffer keyBuf = ByteBuffer.wrap( key );
        keyChannel.write( keyBuf, getOffset(keyOffset2Len));
        writeIndexBlock(block, keyOffset2Len, indexChannel );
    }

    private void writeIndexBlock(CacheBlock<byte[]> block, long keyOffset2Len, FileChannel channel ) throws IOException {
        long pos = OFFSET + block.getRecordNo() * RECORD_SIZE;
        checkFileSize(pos, RECORD_SIZE );
        ByteBuffer buf = ByteBuffer.allocate( RECORD_SIZE);
        buf.put(block.getStatus() );
        buf.putLong(keyOffset2Len );
        buf.putLong(block.getDataOffset2Len() );
        buf.putLong(block.getBlock2Version() );
        buf.putShort(block.getNode() );
        buf.flip();
        channel.write( buf, pos);
    }


    public void writeExistBlock(CacheBlock<byte[]> block) throws IOException {
        ByteBuffer dataBuf = ByteBuffer.wrap( block.getData() );
        // don't use flip() before it write when use wrap
        //dataBuf.flip();
        dataChannel.write( dataBuf, block.getDataOffset() );
        long pos = OFFSET + block.getRecordNo() * RECORD_SIZE;
        int off = 9;
        ByteBuffer buf = ByteBuffer.allocate( RECORD_SIZE - off);
        buf.putLong(block.getDataOffset2Len() );
        buf.putLong(block.getBlock2Version() );
        buf.putShort(block.getNode() );
        // must flip() before it write
        buf.flip();
        indexChannel.write( buf, pos + off);
    }

    public void removeBlock(CacheBlock<byte[]> block) throws IOException{
        long pos = OFFSET + block.getRecordNo() * RECORD_SIZE;
        ByteBuffer buf = ByteBuffer.allocate( 1);
        byte status = (byte) (block.getStatus()  | DELETED ) ;
        buf.put( status );
        buf.flip();
        indexChannel.write( buf, pos);
        writeIndexBlock(block, logOffset, logChannel);
        logOffset += RECORD_SIZE ;
    }

    private void close(FileChannel channel) {
        try {
            if ( channel != null )
                channel.close();
        } catch (IOException ex) {
            logger.error( ex.getMessage(), ex);
            // swallow exception
        }
    }

    public void close() {
        close( indexChannel);
        close( dataChannel);
        close( keyChannel);
        close( logChannel);
    }

    public FileChannel getDataChannel() {
        return dataChannel;
    }

    public FileChannel getKeyChannel() {
        return keyChannel;
    }

    public FileChannel getIndexChannel() {
        return indexChannel;
    }

    public FileChannel getLogChannel() {
        return logChannel;
    }

    public int getTotalRecord() {
        return totalRecord;
    }

    public long getDataOffset() {
        return dataOffset;
    }

    public long getKeyOffset() {
        return keyOffset;
    }

    public long getLogOffset() {
        return logOffset;
    }

    public int getDeleted() {
        return deleted;
    }

    public void setTotalRecord(int totalRecord) {
        this.totalRecord = totalRecord;
    }

    public void setDataOffset(long dataOffset) {
        this.dataOffset = dataOffset;
    }

    public void setKeyOffset(long keyOffset) {
        this.keyOffset = keyOffset;
    }

    public void setLogOffset(long logOffset) {
        this.logOffset = logOffset;
    }

    public void setDeleted(int deleted) {
        this.deleted = deleted;
    }

    public void truncate() throws IOException {
        indexChannel.truncate( OFFSET);
        dataChannel.truncate( OFFSET);
        keyChannel.truncate( OFFSET);
        logChannel.truncate( OFFSET);
    }
}
